package org.apache.spark.doe

import org.apache.spark.rdd.{MapPartitionsRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

object MapPartitionsDemo {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
    val sc = new SparkContext(conf)

    val rdd1: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 2)

//    val rdd2: RDD[Int] = rdd1.mapPartitions(it => {
//      it.map(e => e * 10)
//    })

    //rdd1.mapPartitionsWithIndex()

    val func = (it: Iterator[Int]) => it.map(e => e * 10)

    val rdd2: MapPartitionsRDD[Int, Int] = new MapPartitionsRDD[Int, Int](
      rdd1,
      (_, _, it) => func(it)
    )



    rdd2.saveAsTextFile("out/out03")

  }

}
